package com.vpen.kafkademo.kafka.producer;

import com.vpen.kafkademo.model.entity.StoreContentDownload;
import lombok.extern.slf4j.Slf4j;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.time.LocalDateTime;

/**
 * 生产者
 *
 * @author by Vpen
 * @date 2021/8/24 19:25
 */
@RestController
@RequestMapping("/producer")
@Slf4j
public class KafkaProducer {


    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    private final String TOPIC = "testTopic";
    private final String TOPIC2 = "testTopic2";

    @Value("${kafka.topic}")
    private String topic;


    @GetMapping("/send/{msg}")
    public void send(@PathVariable("msg") String msg) {
        log.info("生产者生成消息:{}", msg);
        kafkaTemplate.send(TOPIC, msg);
    }

    /***
     * 描述: 事务提交
     * @author Vpen
     * @date 2021/8/24 20:33
     * @param msg
     **/
    @GetMapping("/send1/{msg}")
    @Transactional(rollbackFor = {Exception.class})
    public void send1(@PathVariable("msg") String msg) {
        kafkaTemplate.send(TOPIC, msg);
        throw new RuntimeException("发送失败");
    }

    @GetMapping("/upload/downloads")
    @Transactional(rollbackFor = {Exception.class})
    public void uploadDownloads(StoreContentDownload scd)  {

        scd.setTime(LocalDateTime.now());

        kafkaTemplate.send(TOPIC, scd);
    }

    @GetMapping("/upload/downloads2")
    @Transactional(rollbackFor = {Exception.class})
    public void uploadDownloads2(StoreContentDownload scd)  {

        scd.setTime(LocalDateTime.now());
        kafkaTemplate.send(topic, scd);
    }


    @GetMapping("/send2/{msg}")
    @Transactional(rollbackFor = {Exception.class})
    public void send2(@PathVariable("msg") String msg) {

        kafkaTemplate.send(TOPIC, msg);
    }



}
